package job.gfp.ranks.get_ready;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.Set;

import job.gfp.ranks.Customize;
import job.gfp.ranks.base.Data;
import job.gfp.ranks.base.RanksUtil;
import mapred.base.TextMapper;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

import util.FeatureMap;
import util.UrlUtil;

public class GetReady_Mapper extends TextMapper {

    private static double times;

    private static Set<String> seeds;

    @Override
    public void map(Text key, Text value) throws IOException {
        FeatureMap map = FeatureMap.parse(value.toString(), RanksUtil.DELIM);
        long outDegree = 0;
        for (String target : map.featureSet()) {
            outDegree += map.get(target);
        }

        Data data = Customize.initialData(key.toString(), outDegree, seeds,
                times);

        this.output(key, new Text(data + "\t" + value));

        this.incrCounter("WebSize", "WebSize", 1);
    }

    @Override
    public void setUp() {
        times = Double.parseDouble(this.getConfig("SeedsTimes"));
        seeds = new HashSet<String>();

        try {
            BufferedReader br = new BufferedReader(new InputStreamReader(this
                    .getFileSystem().open(new Path(this.getConfig("Seeds")))));
            for (String line = br.readLine(); line != null; line = br
                    .readLine()) {
                seeds.add(UrlUtil.getHost(line.trim()));
            }
            br.close();
            seeds.remove("");
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(255);
        }
    }

    @Override
    public void tearDown() throws IOException {
    }

}
